package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.udfs.GeohashUdf;
import cn.doitedu.rtdw.udfs.Map2JsonUdf;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_行为日志ODS到DWD {
    public static void main(String[] args) {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka的ods层的行为日志
        tenv.executeSql(
                "create table ods_kafka(   \n" +
                "    username string,              \n" +  // 物理字段
                "    release_channel string,       \n" +  // 物理字段
                "    device_type string,           \n" +  // 物理字段
                "    session_id string,            \n" +  // 物理字段
                "    lat double,                   \n" +  // 物理字段
                "    lng double,                   \n" +  // 物理字段
                "    event_id string,              \n" +  // 物理字段
                "    action_time bigint,           \n" +  // 物理字段
                "    properties map<string,string>,\n" +  // 物理字段
                "    pt  as proctime(),            \n" +  // 表达式字段，这是一个带处理时间语义的时间字段  *PROCTIME*
                "    tt  as to_timestamp_ltz(action_time,3) ,            \n" +  // 表达式字段,这只是一个普通的时间字段
                "    rt  as to_timestamp_ltz(action_time,3) ,           \n" +  // 表达式字段,这是一个带事件时间语义的时间字段 *ROWTIME*
                "    watermark for rt as rt - interval '5' second       \n" +  // 声明watermark
                ") with (                          \n" +
                "  'connector' = 'kafka',          \n" +
                "  'topic' = 'ods-user-log',       \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit43-1',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'value.format'='json',                            \n" +
                "  'value.json.fail-on-missing-field'='false',       \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'             \n" +
                ")");

        //tenv.executeSql("desc ods_kafka").print();
        /**
         * |pt | TIMESTAMP_LTZ(3) *PROCTIME* | FALSE |
         * |tt |            TIMESTAMP_LTZ(3) |  TRUE |
         * |rt |  TIMESTAMP_LTZ(3) *ROWTIME* |  TRUE |
         * +-----------------+------------------------
         *
         *
         */


        // 建表，映射hbase中的 用户信息维表
        tenv.executeSql(
                " create table dim_user_info_hbase(       "+
                        " 	user_name string,                      "+
                        "   f  row<                                " +
                        "          id  bigint                      "+
                        " 	       ,member_level_id bigint         "+
                        "          ,password string                "+
                        "          ,nickname string                "+
                        "          ,phone string                   "+
                        "          ,status int                     "+
                        "          ,create_time timestamp(3)       "+
                        "          ,icon string                    "+
                        "          ,gender int                     "+
                        "          ,birthday date                  "+
                        "          ,city string                    "+
                        "          ,job string                     "+
                        "          ,personalized_signature string  "+
                        "          ,source_type int                "+
                        "          ,integration int                "+
                        "          ,growth int                     "+
                        "          ,luckey_count int               "+
                        "          ,history_integration int        "+
                        "          ,modify_time timestamp(3)       "+
                        " 	>                                      "+
                        " ) WITH (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'dim_user_info',         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );


        // 建表，映射hbase中的 地域信息维表
        tenv.executeSql("create table dim_area_info_hbase(              \n" +
                "   geohash    string,                                  \n" +
                "   f ROW<province string,city string, region string>   \n" +
                ") with (                            \n" +
                "'connector' = 'hbase-2.2',          \n" +
                "'table-name' = 'dim_area_info',     \n" +
                "'zookeeper.quorum' = 'doitedu:2181',\n" +
                "'lookup.cache' = 'PARTIAL',         \n" +
                "'lookup.partial-cache.max-rows' = '30000'     \n" +

                ")");


        // 建表，映射hbase中的 页面信息维表
        tenv.executeSql(
                "create table dim_page_info_hbase(                \n" +
                "   url_prefix    string,                         \n" +
                "   f ROW<page_type string,page_service string>   \n" +
                ") with (                            \n" +
                "'connector' = 'hbase-2.2',          \n" +
                "'table-name' = 'dim_page_info',     \n" +
                "'zookeeper.quorum' = 'doitedu:2181',\n" +
                "'lookup.cache' = 'PARTIAL',         \n" +
                "'lookup.partial-cache.max-rows' = '1000'     \n" +


                ")");





        // 建表，映射kafka的dwd层的行为日志
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "     release_channel string                              "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,luckey_count int                                   "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "


        );


        // 建表，映射doris中的dwd层的明细表
        tenv.executeSql(
                "create table dwd_event_detail_doris(\n" +
                        "    province               varchar,\n" +
                        "    city                   varchar,\n" +
                        "    region                 varchar,\n" +
                        "    user_id                bigint,\n" +
                        "    username               varchar,\n" +
                        "    session_id             varchar,\n" +
                        "    release_channel        varchar,\n" +
                        "    device_type            varchar,\n" +
                        "    lat                    double,\n" +
                        "    lng                    double,\n" +
                        "    eventId                varchar,\n" +
                        "    action_time            bigint,\n" +
                        "    properties             varchar,\n" +
                        "    member_level_id        bigint,\n" +
                        "    password               varchar,\n" +
                        "    nickname               varchar,\n" +
                        "    phone                  varchar,\n" +
                        "    status                 int,\n" +
                        "    create_time            timestamp(3),\n" +
                        "    icon                   varchar,\n" +
                        "    gender                 int,\n" +
                        "    birthday               date,\n" +
                        "    register_city          varchar,\n" +
                        "    job                    varchar,\n" +
                        "    personalized_signature varchar,\n" +
                        "    source_type            int,\n" +
                        "    integration            int,\n" +
                        "    growth                 int,\n" +
                        "    luckey_count           int,\n" +
                        "    history_integration    int,\n" +
                        "    modify_time            timestamp(3),\n" +
                        "    page_type              varchar,\n" +
                        "    page_service           varchar,\n" +
                        "    dt                     date\n" +
                        ")  with (                                           \n" +
                        "    'connector' = 'doris',                          \n" +
                        "    'fenodes' = 'doitedu:8030',                     \n" +
                        "    'table.identifier' = 'dwd.user_events_detail',  \n" +
                        "    'username' = 'root',                            \n" +
                        "    'password' = 'root',                            \n" +
                        "    'sink.label-prefix' = 'doit42_no1'              \n" +
                        ")                                                   "
        );




        // insert into ... select  join
        tenv.createTemporaryFunction("geo", GeohashUdf.class);
        tenv.executeSql(
                " create temporary view joined_result AS                                                                                         "+
                        " select   /*+ LOOKUP('table'='dim_user_info_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */      "+
                        "      t1.release_channel                                                                                                        "+
                        "     ,t1.device_type                                                                                                            "+
                        "     ,t1.session_id                                                                                                             "+
                        "     ,t1.lat                                                                                                                    "+
                        "     ,t1.lng                                                                                                                    "+
                        "     ,t1.username                                                                                                               "+
                        "     ,t1.event_id                                                                                                               "+
                        "     ,t1.action_time                                                                                                            "+
                        "     ,t1.properties                                                                                                             "+
                        "     ,t2.f.id as user_id                                                                                                          "+
                        "     ,t2.f.member_level_id                                                                                                        "+
                        "     ,t2.f.password                                                                                                               "+
                        "     ,t2.f.nickname                                                                                                               "+
                        "     ,t2.f.phone                                                                                                                  "+
                        "     ,t2.f.status                                                                                                                 "+
                        "     ,t2.f.create_time                                                                                                            "+
                        "     ,t2.f.icon                                                                                                                   "+
                        "     ,t2.f.gender                                                                                                                 "+
                        "     ,t2.f.birthday                                                                                                               "+
                        "     ,t2.f.city  as register_city                                                                                                 "+
                        "     ,t2.f.job                                                                                                                    "+
                        "     ,t2.f.personalized_signature                                                                                                 "+
                        "     ,t2.f.source_type                                                                                                            "+
                        "     ,t2.f.integration                                                                                                            "+
                        "     ,t2.f.growth                                                                                                                 "+
                        "     ,t2.f.luckey_count                                                                                                           "+
                        "     ,t2.f.history_integration                                                                                                    "+
                        "     ,t2.f.modify_time                                                                                                            "+
                        "     ,t3.f.province                                                                                                               "+
                        "     ,t3.f.city                                                                                                                   "+
                        "     ,t3.f.region                                                                                                                 "+
                        "     ,t4.f.page_type                                                                                                              "+
                        "     ,t4.f.page_service                                                                                                           "+
                        " from ods_kafka t1                                                                                                                "+
                        " LEFT JOIN dim_user_info_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t2 ON t1.username=t2.user_name                                   "+
                        " LEFT JOIN dim_area_info_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t3 ON geo(t1.lat,t1.lng)=t3.geohash                             "+
                        " LEFT JOIN dim_page_info_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t4 ON regexp_extract(t1.properties['url'],'(^.*/).*?')=t4.url_prefix   "
        );


        // 将join结果，插入kafka明细层映射表
        tenv.executeSql("insert into dwd_events_kafka select  * from joined_result");


        // 将join结果，插入doris明细层映射表
        tenv.createTemporaryFunction("map2json", Map2JsonUdf.class);
        tenv.executeSql("INSERT INTO dwd_event_detail_doris SELECT                                \n" +
                "    province                                                                      \n" +
                "    ,city                                                                         \n" +
                "    ,region                                                                       \n" +
                "    ,user_id                                                                      \n" +
                "    ,username                                                                     \n" +
                "    ,session_id                                                                   \n" +
                "    ,release_channel                                                              \n" +
                "    ,device_type                                                                  \n" +
                "    ,lat                                                                          \n" +
                "    ,lng                                                                          \n" +
                "    ,event_id                                                                      \n" +
                "    ,action_time                                                                  \n" +
                "    ,map2json(properties) as properties                                           \n" +
                "    ,member_level_id                                                              \n" +
                "    ,password                                                                     \n" +
                "    ,nickname                                                                     \n" +
                "    ,phone                                                                        \n" +
                "    ,status                                                                       \n" +
                "    ,create_time                                                                  \n" +
                "    ,icon                                                                         \n" +
                "    ,gender                                                                       \n" +
                "    ,birthday                                                                     \n" +
                "    ,register_city                                                                \n" +
                "    ,job                                                                          \n" +
                "    ,personalized_signature                                                       \n" +
                "    ,source_type                                                                  \n" +
                "    ,integration                                                                  \n" +
                "    ,growth                                                                       \n" +
                "    ,luckey_count                                                                 \n" +
                "    ,history_integration                                                          \n" +
                "    ,modify_time                                                                  \n" +
                "    ,page_type                                                                    \n" +
                "    ,page_service                                                                 \n" +
                "    ,to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd'))  dt            \n" +
                "FROM joined_result                                                                ");




    }
}
